package com.aotain.common.mongo;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map.Entry;
import java.util.Set;

import net.sf.json.JSONArray;
import net.sf.json.JSONObject;

import org.apache.log4j.Logger;
import org.bson.Document;
import org.bson.conversions.Bson;

import com.aotain.common.ZkKafkaConfig;
import com.mongodb.BasicDBObject;
import com.mongodb.DBObject;
import com.mongodb.MongoClient;
import com.mongodb.MongoClientOptions;
import com.mongodb.MongoException;
import com.mongodb.ServerAddress;
import com.mongodb.client.FindIterable;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.Sorts;
import com.mongodb.client.model.UpdateOptions;
import com.mongodb.client.result.UpdateResult;


/**
 * 
 * mongodb连接
 * <功能详细描述>
 * 
 * @author  Turk
 * @version  [版本号, 2017年4月20日]
 * @see  [相关类/方法]
 * @since  [产品/模块版本]
 */
public class MongoConnection {

	private static MongoConnection instance = null;
	MongoClient mongoClient = null;
	private String mongodbIP = "";
	private int mongodbPort = 27017;
	private long Counter = 0;

	public static MongoConnection getInstance() {
		if(instance == null)
			instance = new MongoConnection();
		return instance;		
	}

	public MongoConnection() {
		// 获取链接    
		OpenConnection();
	}
	
	public MongoConnection(String ip,int port) {
		OpenConnection(ip,port);
	}

	/**
	 * 
	 * 打开数据库
	 * <功能详细描述>
	 * @see [类、类#方法、类#成员]
	 */
	private void OpenConnection() {
		if(mongoClient == null) {
			ZkKafkaConfig z_kConfig = new ZkKafkaConfig("/home/storm/config/dbconfig.ini");
//			ZkKafkaConfig z_kConfig = new ZkKafkaConfig("E:\\dbconfig.ini");
			if(z_kConfig.getMongoDBIP()!=null) {
				mongoClient = new MongoClient(z_kConfig.getMongoDBIP(), z_kConfig.getMongoDBPort());
			} else {
				mongoClient = new MongoClient("192.168.5.98", 27017);
				Logger.getLogger(MongoConnection.class).error("Config Error:MongoDB IP is null.");
			}
			
			Logger.getLogger(MongoConnection.class).info("MongoDB Open Num:" + Counter++);
		}
	}
	
	/**
	 * 
	 * 打开数据库
	 * <功能详细描述>
	 * @see [类、类#方法、类#成员]
	 */
	private void OpenConnection(String ip,int port) {
//		if(mongoClient == null) {
//			mongoClient = new MongoClient(ip, port);
//		}
		
		if(mongoClient == null){
            MongoClientOptions.Builder build = new MongoClientOptions.Builder();        
            build.connectionsPerHost(10);   //与目标数据库能够建立的最大connection数量为50
//            build.autoConnectRetry(true);   //自动重连数据库启动
            build.threadsAllowedToBlockForConnectionMultiplier(50); //如果当前所有的connection都在使用中，则每个connection上可以有50个线程排队等待
            /*
             * 一个线程访问数据库的时候，在成功获取到一个可用数据库连接之前的最长等待时间为2分钟
             * 这里比较危险，如果超过maxWaitTime都没有获取到这个连接的话，该线程就会抛出Exception
             * 故这里设置的maxWaitTime应该足够大，以免由于排队线程过多造成的数据库访问失败
             */
            build.maxWaitTime(1000*60*2);
            build.connectTimeout(1000*60*1);    //与数据库建立连接的timeout设置为1分钟
            

            MongoClientOptions myOptions = build.build();  
            
            ServerAddress sa = new ServerAddress(ip, port);
            
            try {
                //数据库连接实例
                mongoClient = new MongoClient(sa, myOptions);
            } catch (MongoException e){
                e.printStackTrace();
            }

        }
		
	}

	/**
	 * 
	 * 向MongoDB 写入数据
	 * <功能详细描述>
	 * @param dbname 数据库名
	 * @param docname 集合名
	 * @param data 数据集
	 * @see [类、类#方法、类#成员]
	 */
	public void InsertData(String dbname,String collection,List<List<DataColumn>> data) {

		OpenConnection();

		// 获取数据库    
		MongoDatabase database = mongoClient.getDatabase(dbname);    
		// 进入某个文档集(表)    
		MongoCollection<Document>  cols = database.getCollection(collection);

		for(List<DataColumn> row : data) {

			Document doc = new Document();
			for(DataColumn column : row) {
				doc.put(column.getKey(), column.getValue());
			}
			cols.insertOne(doc);
		}
		//CloseConnection();

	}
	
	
	/**
	 * 更新MongoDB中的数据
	 * @param dbname
	 * @param collection
	 * @param rowkey
	 * @param rowdata
	 */
	public UpdateResult UpdateData(String dbname,String collection,List<DataColumn> filter, List<DataColumn> rowdata) {
		OpenConnection();
		
		// 获取数据库    
		MongoDatabase database = mongoClient.getDatabase(dbname);    
		// 进入某个文档集(表)    
		MongoCollection<Document>  cols = database.getCollection(collection);
			
		List<Bson> filterlist = new ArrayList<Bson>();
		for(DataColumn column : filter){
			Bson bfilter = Filters.eq(column.getKey(), column.getValue());
			filterlist.add(bfilter);
		}
		Bson andfilter = Filters.and(filterlist);
		Document newdoc = new Document();
		
		for(DataColumn column : rowdata) {
			newdoc.put(column.getKey(), column.getValue());
		}
		 
		UpdateOptions updateOp = new UpdateOptions();
		updateOp.upsert(true);//如果不存在，插入新值
		return cols.updateOne(andfilter, new Document("$set", newdoc), updateOp);
	}
	
	
	/**
	 * 更新MongoDB中的数据
	 * @param dbname
	 * @param collection
	 * @param rowkey
	 * @param rowdata
	 */
	public UpdateResult PushData(String dbname,String collection,List<DataColumn> filter, List<DataColumn> rowdata) {
		OpenConnection();
		
		// 获取数据库    
		MongoDatabase database = mongoClient.getDatabase(dbname);    
		// 进入某个文档集(表)    
		MongoCollection<Document>  cols = database.getCollection(collection);
			
		List<Bson> filterlist = new ArrayList<Bson>();
		for(DataColumn column : filter){
			Bson bfilter = Filters.eq(column.getKey(), column.getValue());
			filterlist.add(bfilter);
		}
		Bson andfilter = Filters.and(filterlist);
		Document newdoc = new Document();
		
		for(DataColumn column : rowdata) {
			newdoc.put(column.getKey(), column.getValue());
		}
		 
		UpdateOptions updateOp = new UpdateOptions();
		updateOp.upsert(true);//如果不存在，插入新值
		return cols.updateOne(andfilter, new Document("$push", newdoc), updateOp);
	}
	
	/**
	 * 更新MongoDB中的数据,数值增量累加
	 * @param dbname
	 * @param collection
	 * @param rowkey
	 * @param rowdata
	 */
	public UpdateResult IncData(String dbname,String collection,List<DataColumn> filter, List<DataColumn> rowdata) {
		OpenConnection();
		
		// 获取数据库    
		MongoDatabase database = mongoClient.getDatabase(dbname);    
		// 进入某个文档集(表)    
		MongoCollection<Document>  cols = database.getCollection(collection);
			
		List<Bson> filterlist = new ArrayList<Bson>();
		for(DataColumn column : filter){
			Bson bfilter = Filters.eq(column.getKey(), column.getValue());
			filterlist.add(bfilter);
		}
		Bson andfilter = Filters.and(filterlist);
		Document newdoc = new Document();
		for(DataColumn column : rowdata) {
			newdoc.put(column.getKey(), column.getValue());
		}
		 
		UpdateOptions updateOp = new UpdateOptions();
		updateOp.upsert(true);//如果不存在，插入新值
		return cols.updateOne(andfilter, new Document("$inc", newdoc), updateOp);
	}
	
	/**
	 * 更新MongoDB中的数据
	 * @param dbname
	 * @param collection
	 * @param rowkey
	 * @param jsonData
	 */
	public UpdateResult UpdateDataByJson(String dbname,String collection,List<DataColumn> filter, String jsonData) {
		try {
			OpenConnection();
			
			// 获取数据库    
			MongoDatabase database = mongoClient.getDatabase(dbname);    
			// 进入某个文档集(表)    
			MongoCollection<Document>  cols = database.getCollection(collection);
			
			List<Bson> filterlist = new ArrayList<Bson>();
			for(DataColumn column : filter){
				Bson bfilter = Filters.eq(column.getKey(), column.getValue());
				filterlist.add(bfilter);
			}
			Bson andfilter = Filters.and(filterlist);
			
			Document newdoc = Document.parse(jsonData);
	
			 
			UpdateOptions updateOp = new UpdateOptions();
			updateOp.upsert(true);//如果不存在，插入新值
			return cols.updateOne(andfilter, new Document("$set", newdoc), updateOp);
		} catch(Exception ex) {
			Logger.getLogger(MongoConnection.class).error("JSON:" + jsonData);
			ex.printStackTrace();
			return null;
		}
	}
	
	/**
	 * 更新MongoDB中的数据，addToSet避免重复数组
	 * @param dbname
	 * @param collection
	 * @param rowkey
	 * @param jsonData
	 */
	public UpdateResult PushDataByJson(String dbname,String collection,List<DataColumn> filter, String jsonData) {
		try {
			OpenConnection();
			
			// 获取数据库    
			MongoDatabase database = mongoClient.getDatabase(dbname); 
			
			// 进入某个文档集(表)    
			MongoCollection<Document>  cols = database.getCollection(collection);
			
			List<Bson> filterlist = new ArrayList<Bson>();
			for(DataColumn column : filter){
				Bson bfilter = Filters.eq(column.getKey(), column.getValue());
				filterlist.add(bfilter);
			}
			Bson andfilter = Filters.and(filterlist);
			
			Document newdoc = Document.parse(jsonData);
	
			 
			UpdateOptions updateOp = new UpdateOptions();
		
			updateOp.upsert(true);//如果不存在，插入新值
			
			return cols.updateOne(andfilter, new Document("$addToSet", newdoc), updateOp);
		} catch(Exception ex) {
			Logger.getLogger(MongoConnection.class).error("JSON:" + jsonData,ex);
			//ex.printStackTrace();
			return null;
		}
	}
	
	
	/**
	 * 更新MongoDB中的数据
	 * @param dbname
	 * @param collection
	 * @param rowkey
	 * @param jsonData
	 */
	public void PullDataByJson(String dbname,String collection,List<DataColumn> filter, String jsonData) {
		try {
			OpenConnection();
			
			// 获取数据库    
			MongoDatabase database = mongoClient.getDatabase(dbname); 
			
			// 进入某个文档集(表)    
			MongoCollection<Document>  cols = database.getCollection(collection);
			
			List<Bson> filterlist = new ArrayList<Bson>();
			for(DataColumn column : filter){
				Bson bfilter = Filters.eq(column.getKey(), column.getValue());
				filterlist.add(bfilter);
			}
			Bson andfilter = Filters.and(filterlist);
			
			Document newdoc = Document.parse(jsonData);
	
			 
			UpdateOptions updateOp = new UpdateOptions();
		
			updateOp.upsert(true);//如果不存在，插入新值
			
			cols.updateOne(andfilter, new Document("$pull", newdoc), updateOp);
		} catch(Exception ex) {
			Logger.getLogger(MongoConnection.class).error("JSON:" + jsonData,ex);
			//ex.printStackTrace();
		}
	}



	/**
	 * 根据条件查询
	 * @Title: getResult 
	 * @Description: TODO(这里用一句话描述这个方法的作用) 
	 * @param dbName
	 * @param tableName
	 * @param limit
	 * @param starttime
	 * @param endtime
	 * @param conditions 字段条件集合
	 * @return
	 * @return List<String>    返回类型 
	 * @author 程彬
	 * @date 2017年4月26日 下午5:48:24
	 */
	public String getResult(String dbName,String tableName,int limit,String starttime,String endtime,HashMap<String,Object> conditions,String sort) {

		OpenConnection();
		MongoDatabase database = mongoClient.getDatabase(dbName);
		MongoCollection<Document> collection = database.getCollection(tableName);

		//#### 条件字段(单个或多个)过滤器
		BasicDBObject docfilter = new BasicDBObject();
		if(conditions != null ) {
			for(Entry<String,Object> entry : conditions.entrySet()) {
				//				filter = filter.
				String key = entry.getKey();
				Object val = entry.getValue();
				docfilter.append(key, val);
			}
		}

		//#### 日期范围过滤器
		Bson datefilter = null;
		if(starttime != null && endtime == null) {
			datefilter = Filters.and(Filters.gte("ACCESSTIME", Long.parseLong(starttime)));
		} else if (starttime != null && endtime != null) {
			datefilter = Filters.and(Filters.gte("ACCESSTIME", Long.parseLong(starttime)),Filters.lte("ACCESSTIME", Long.parseLong(endtime)));;
		} else if (starttime == null && endtime != null) {
			datefilter = Filters.and(Filters.lte("ACCESSTIME", Long.parseLong(endtime)));
		}

		Bson resfilter = null;
		if(datefilter != null && docfilter.size()>0) {
			resfilter = Filters.and(docfilter,datefilter);
		} else if(datefilter == null && docfilter.size() > 0) {
			resfilter = Filters.and(docfilter);
		} else if(datefilter != null && docfilter.size()== 0) {
			resfilter = Filters.and(datefilter);
		}

		List<String> reslist = new ArrayList<String>();
		FindIterable<Document> iter;
		if(resfilter != null) {
			if(sort != null) {
				iter = collection.find(resfilter,Document.class).limit(limit).sort(Sorts.descending(sort));
			}else {
				iter = collection.find(resfilter,Document.class).limit(limit).sort(Sorts.descending("REPORTTIME"));
			}
		} else {
			if(sort != null) {
				iter = collection.find(Document.class).limit(limit).sort(Sorts.descending(sort));
			}else {
				iter = collection.find(Document.class).limit(limit).sort(Sorts.descending("REPORTTIME"));
			}
		}
		MongoCursor<Document> cursor = iter.iterator();

		while(cursor.hasNext()) {
			Document resdoc = cursor.next();
			reslist.add(resdoc.toJson());
		}

		return parseReult(reslist);
	}

	/**
	 * 格式化查询结果
	 * @Title: parseReult 
	 * @Description: TODO(这里用一句话描述这个方法的作用) 
	 * @param reslist
	 * @return
	 * @return String    返回类型 
	 * @author 程彬
	 * @date 2017年4月27日 下午4:22:01
	 */
	private String parseReult(List<String> reslist) {
		/**
		 * [{"_id":{"$oid":"590150a0634ab5573156a0da"},"SIP":"192.168.8.245","REPORTTIME":{"$numberLong":"201704271000"},"DNSTREAMPACKET":{"$numberLong":"26948"},"DNSTREAMOCTETS":{"$numberLong":"35947517"},"UPSTREAMOCTETS":{"$numberLong":"15166889"},"UPSTREAMPACKET":{"$numberLong":"42398"}},
		 *  {"_id":{"$oid":"59015eb0634ab5573156b3c6"},"SIP":"192.168.8.245","REPORTTIME":{"$numberLong":"201704271100"},"DNSTREAMPACKET":{"$numberLong":"19068"},"DNSTREAMOCTETS":{"$numberLong":"19513946"},"UPSTREAMOCTETS":{"$numberLong":"2977934"},"UPSTREAMPACKET":{"$numberLong":"19554"}}]
		 *	格式清理，去掉"$numberLong" 等
		 */
		JSONArray resultJO = new JSONArray();
		JSONArray jArr = JSONArray.fromObject(reslist);
		Iterator<Object> jsonIter = jArr.iterator();

		while(jsonIter.hasNext()) {

			JSONObject obj = (JSONObject) jsonIter.next();		
			Iterator<Object> innerIter = obj.keys();
			while(innerIter.hasNext()) {

				String key = (String) innerIter.next();
				Object next = obj.get(key);
				if(next instanceof JSONObject) {

					JSONObject innerObj = (JSONObject) next;
					Iterator<Object> iter3 = innerObj.keys();
					while(iter3.hasNext()) {

						String key3 = (String) iter3.next();
						String value = (String) innerObj.get(key3);
						obj.element(key, value);
					}
				}
			}

			resultJO.element(obj);

		}
		String resStr = resultJO.toString();
		return resStr;
	}
	
	public List<ArrayList<DataColumn>> getResultRow(String dbName,String tableName, List<FilterItem> filterList) {
		List<ArrayList<DataColumn>> returnData = new ArrayList<ArrayList<DataColumn>>();
		
		OpenConnection();
		MongoDatabase database = mongoClient.getDatabase(dbName);
		MongoCollection<Document> collection = database.getCollection(tableName);

		Bson resfilter = null;
		for(FilterItem filter : filterList) {
			Bson item = null;
			if(filter.getCondition() == FilterItem.Condition.EQ) {
				item = Filters.and(Filters.eq(filter.getColumnName(), filter.getColumnValue()));
			} else if (filter.getCondition() == FilterItem.Condition.GR) {
				item = Filters.and(Filters.gt(filter.getColumnName(), filter.getColumnValue()));
			} else if (filter.getCondition() == FilterItem.Condition.LE) {
				item = Filters.and(Filters.lt(filter.getColumnName(), filter.getColumnValue()));
			} else if (filter.getCondition() == FilterItem.Condition.EQGR) {
				item = Filters.and(Filters.gte(filter.getColumnName(), filter.getColumnValue()));
			} else if (filter.getCondition() == FilterItem.Condition.EQLE) {
				item = Filters.and(Filters.lte(filter.getColumnName(), filter.getColumnValue()));
			}
			
			if(resfilter!=null) {
				resfilter = Filters.and(resfilter,item);
			} else {
				resfilter = Filters.and(item);
			}
			
		}
		
		if(resfilter != null) {
			FindIterable<Document> iter = collection.find(resfilter,Document.class);
			MongoCursor<Document> cursor = iter.iterator();
			while(cursor.hasNext()) {
				Document resdoc = cursor.next();
				ArrayList<DataColumn> row = new ArrayList<DataColumn>();
				for(Entry<String, Object> entry : resdoc.entrySet()) {
					DataColumn column = new DataColumn();
					column.setKey(entry.getKey());
					column.setValue(entry.getValue());
					row.add(column);
				}
				returnData.add(row);
			}
		}
//		CloseConnection();
		return returnData;
	}

		

	public void CloseConnection() {
		mongoClient.close();
		mongoClient = null;
	}

	public static void main(String[] argv){ 
		MongoConnection dbconn = new MongoConnection("192.168.5.95",27017);
		//		List<List<DataColumn>> data  = new ArrayList<List<DataColumn>>();
		//		for(int i=0;i<10;i++){
		//			List<DataColumn> row = new ArrayList<DataColumn>();
		//			for(int j=0;j<10;j++){
		//				DataColumn column = new DataColumn();
		//				column.setKey("key" + i + "-" + j);
		//				column.setValue("value" + i + "-" + j );
		//				row.add(column);
		//			}
		//			data.add(row);
		//		}
		//		dbconn.InsertData("sds", "test", data);
		//		System.out.println("Insert Success!");
		//		System.out.println(dbconn.getResult("sds", "METIS_FLOW_STAT_30MIN"));
		//		String s1 = "{ \"_id\" : ObjectId(\"58ffe300634ab5573155703e\")}";
		//		String s2 = "{ \"_name\" : ObjectId(\"58ffe300634ab5573155703e\")}";
		//		List<String> list = new ArrayList<String>();
		//		list.add(s1);
		//		list.add(s2);
		//		System.out.println(JSONArray.fromObject(list).toString());
//		HashMap<String,Object> cond = new HashMap<String,Object>();
//		cond.put("DESTIP", "121.10.255.20");
//		cond.put("ABRNORMAL", 4);
//		String res = dbconn.getResult("sds", "SDS_ABNORMAL_LOG2", 1000, "20170809094300", "20170809094400", cond,"ACCESSTIME");
//		System.out.println(res);
//		String tableName = "METIS_ABNORMAL_LOG";
//		List<DataColumn> row = new ArrayList<DataColumn>();
//		List<DataColumn> filter = new ArrayList<DataColumn>();
//		row.add(new DataColumn("EVALUATE","80"));
//		filter.add(new DataColumn("ROWKEY","20170509171100_192.168.3.149_183.3.225.58_3"));
//		dbconn.UpdateData("sds", tableName, filter, row);
//		HashMap<String,String> cond = new HashMap<String,String>();
//		cond.put("SIP", "192.168.8.245");
//		String res = dbconn.getResult("sds", "METIS_FLOW_STAT_HOUR", 1000, "201704271000", "201704271800", cond,"REPORTTIME");
//		System.out.println(res);
		String tableName = "TW_USER_TAG_MAIN";
		List<ArrayList<DataColumn>> data = new ArrayList<ArrayList<DataColumn>>();
		
		List<FilterItem> filter = new ArrayList<FilterItem>();
		
		filter.add(new FilterItem("userid","1030401",FilterItem.Condition.EQ));
		filter.add(new FilterItem("20000_3.tagdesc","23333",FilterItem.Condition.EQ));
		data = dbconn.getResultRow("iptv", tableName, filter);
		
		for(ArrayList<DataColumn> row : data) {
			for(DataColumn col : row){
				System.out.println(col.getKey() + ":" + col.getValue());
			}
		}
	}

}
